// 查询表的列
// 主键
// 非空唯一索引
// 组装列名排序
// 如果是大字段类型:Long\CLOB\NCLOB\XMLType\LongRaw\BLOB\BFILE,则排除
// 排序分页查询:
// 11g: rownum
// 12c: OFFSET FETCH
// 使用查询缓存

use std::{
    fmt::{Debug, Display},
    process::exit,
};

use chrono::Local;
use oracle::{
    pool::{self, PoolBuilder},
    Row, RowValue, Version,
};
use tdc::{Action, ColumnInfo, RuleValue, TableInfo};
// use tdc::{Checksum, ColumnInfo, SchemaInfo, TableColumn, TableInfo};

// macro_rules! is_regex {
//     ($s:expr) => {
//         ($s.trim().starts_with("/") || $s.trim().starts_with("!/")) && $s.trim().ends_with("/")
//     };
// }

#[derive(Debug, Default, Clone, RowValue)]
pub struct QueryTable {
    // 所属用户
    pub owner: String,
    // 表名
    pub table_name: String,
}

impl QueryTable {
    pub fn new(owner: String, table_name: String) -> Self {
        QueryTable { owner, table_name }
    }
}

#[derive(Debug, Default, Clone, RowValue)]
pub struct QueryTableColumn {
    pub column_name: String,
    pub data_type: String,
    #[row_value(with = "parse_bool")]
    pub nullable: bool,
}

fn parse_bool(row: &Row, column_name: &str) -> oracle::Result<bool> {
    let value: String = row.get(column_name).unwrap();
    if value == "Y" || value.starts_with("Y") {
        Ok(true)
    } else {
        Ok(false)
    }
}

// 判断是否为自定义正则表达式
// fn is_valid_regex(s: &str) -> bool {
//     s.starts_with("/") && s.ends_with("/")
// }

#[derive(Debug, Clone)]
pub struct OracleManager {
    pool: oracle::pool::Pool,
    dialect: OracleDialect,
}

// const ALL_TAB_COLUMNS: &str = "select COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE from ALL_TAB_COLUMNS where OWNER=:1 and TABLE_NAME=:2 order by COLUMN_ID";
const TAB_CONSTRAINT_PK: &str = "select B.COLUMN_NAME from ALL_CONSTRAINTS A left join ALL_CONS_COLUMNS B ON A.OWNER = B.OWNER and A.TABLE_NAME = B.TABLE_NAME and A.CONSTRAINT_NAME = B.CONSTRAINT_NAME where A.OWNER = :1 and A.TABLE_NAME = :2 and A.CONSTRAINT_TYPE = 'P' order by B.POSITION";
const TAB_CONSTRAINT_UI: &str = "select INDEX_NAME, LISTAGG(COLUMN_NAME, ',') WITHIN GROUP (order by COLUMN_POSITION) COLUMN_NAME from (select A.INDEX_NAME, b.COLUMN_NAME, b.COLUMN_POSITION from ALL_INDEXES A left join ALL_IND_COLUMNS B ON A.TABLE_OWNER = B.TABLE_OWNER and A.TABLE_NAME = B.TABLE_NAME and A.INDEX_NAME = B.INDEX_NAME where A.TABLE_OWNER = :1 and A.TABLE_NAME = :2 and A.UNIQUENESS = 'UNIQUE' ORDER BY B.COLUMN_POSITION) T group by INDEX_NAME";

// 分页查询
// pub fn get_paging_sql(sql: &str) -> String {
//     log::info!("!!!!!!!!!!!!!!!{}", paging!("select count(*) from {}", "abc"));
//     format!(
//         "SELECT a.* FROM (SELECT ROWID AS \"$ROWID\", t.*, ROWNUM AS rn FROM ( {sql} ) t WHERE ROWNUM <= :1 ) a WHERE rn >= :2"
//     )
// }

#[derive(Debug, Clone)]
pub struct OracleDialect {}

const ALL_TABLES_WITH_OWNER: &str =
    "select TABLE_NAME from ALL_TABLES where OWNER = :1 order by TABLE_NAME";
const ALL_TABLES: &str = "select TABLE_NAME from ALL_TABLES";
const ALL_USERS: &str = "select USERNAME SCHEMA_NAME from ALL_USERS";
const ALL_COLUMNS: &str =
    "select COLUMN_NAME, DATA_TYPE, DATA_LENGTH, NULLABLE from ALL_TAB_COLUMNS";
const NLS_SESSION_PARAMETERS: &str =
    "SELECT PARAMETER,VALUE FROM NLS_SESSION_PARAMETERS WHERE PARAMETER LIKE 'NLS%'";

impl OracleDialect {
    pub fn new() -> Self {
        OracleDialect {}
    }

    // 查询所有用户名
    pub fn schemas(&self, values: &Vec<RuleValue>) -> (String, Vec<String>) {
        // let placeholders = values
        //     .iter()
        //     .enumerate()
        //     .map(|(i, _)| format!(":{}", i + 1))
        //     .collect::<Vec<_>>();
        let mut sql = String::from(ALL_USERS);
        // sql.push_str(" where ");

        // let mut in_values = vec![];
        // let mut not_in_values = vec![];

        // // in
        // for (i, value) in values.iter().enumerate() {
        //     if value.re {
        //         continue;
        //     }
        //     if !value.not {
        //         in_values.push(format!(":{}", i + 1));
        //     }
        // }

        // // 全匹配优先级最高
        // for (i, value) in values.iter().enumerate() {
        //     if value.re {
        //         sql.push_str(&format!(
        //             " and {} REGEXP_LIKE (USERNAME, :{})",
        //             if value.not { "not" } else { "" },
        //             i + 1
        //         ));
        //     } else {
        //         sql.push_str(&format!(
        //             " and USERNAME {} in (:{})",
        //             if value.not { "not" } else { "" },
        //             i + 1
        //         ));
        //     }
        // }
        // format!(
        //     "{ALL_USERS} where USERNAME IN ({}) order by USERNAME",
        //     placeholders.join(",")
        // )
        // log::debug!("schemas::sql: {sql}");
        let key = "USERNAME";
        let mut in_values = vec![];
        let mut not_in_values = vec![];
        let mut params = Vec::new();
        for v in values {
            if v.re {
                continue;
            }
            params.push(v.value.clone());
            if !v.not {
                in_values.push(v.value.clone());
            } else {
                not_in_values.push(v.value.clone());
            }
        }

        let cond_in_values = if in_values.len() > 0 {
            format!(
                " and {key} in ({})",
                in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + 1))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        let cond_not_in_values = if not_in_values.len() > 0 {
            format!(
                " and {key} not in ({})",
                not_in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + in_values.len() + 1))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        // 模糊匹配
        let mut regex_likes = String::new();
        let mut not_regex_likes = String::new();
        let mut start_index = in_values.len() + not_in_values.len() + 1;
        for v in values {
            if !v.re {
                continue;
            }
            if !v.not {
                params.push(v.value.clone());
                regex_likes.push_str(&format!(" and REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }

        for v in values {
            if !v.re {
                continue;
            }
            if v.not {
                params.push(v.value.clone());
                not_regex_likes
                    .push_str(&format!(" and NOT REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }

        let sql_condition = format!(
            " where 1 = 1 {cond_in_values} {cond_not_in_values} {regex_likes} {not_regex_likes} order by {key}"
        );

        // let sql_params: Vec<&dyn oracle::sql_type::ToSql> = params
        //     .iter()
        //     .map(|k| k as &dyn oracle::sql_type::ToSql)
        //     .collect();

        sql.push_str(&sql_condition);

        (sql, params)
    }

    pub fn all_schemas(&self) -> String {
        ALL_USERS.to_owned()
    }

    // 查询指定表，不分页查询
    // -- 优先级: 包含全匹配 > 包含模糊匹配 > 排除全匹配 > 排除模糊匹配
    pub fn tables(&self, s: &str, values: &Vec<RuleValue>) -> (String, Vec<String>) {
        // let sql_prefix = "SELECT TABLE_NAME FROM ALL_TABLES WHERE OWNER = :1";

        // if in_values.len() > 0 {
        //     // let placeholders = in_values
        //     //     .iter()
        //     //     .enumerate()
        //     //     .map(|(i, value)| !(value.starts_with("/") && value.ends_with("/") ) {format!(":{}", i + 2)})
        //     //     .collect::<Vec<_>>();

        //     // -- 优先级: 包含全匹配 > 包含模糊匹配 > 排除全匹配 > 排除模糊匹配
        //     // "LBK1=>[TAB1,TTT_*,/TEST*/]"  => LBK1=>[TAB1,TTT_*,/TEST*/]
        //     // "LBK1=>[!TAB1,/TEST*/]"       => LBK1=>[/TEST*/]

        //     let mut placeholders = Vec::new();
        //     let mut index = 2;
        //     for value in in_values {
        //         if is_regex!(value) {
        //             continue;
        //         }
        //         placeholders.push(format!(":{}", index));
        //         index += 1;
        //     }
        //     let cond_in = if placeholders.len() > 0 {
        //         format!("TABLE_NAME IN ({})", placeholders.join(","))
        //     } else {
        //         String::new()
        //     };

        //     placeholders.clear();
        //     for value in in_values {
        //         if !is_regex!(value) {
        //             continue;
        //         }
        //         placeholders.push(format!("REGEXP_LIKE (TABLE_NAME, :{})", index));
        //         index += 1;
        //     }

        //     let cond_or = if placeholders.len() > 0 {
        //         format!("{}", placeholders.join(" OR "))
        //     } else {
        //         String::new()
        //     };

        //     if cond_in.len() > 0 || cond_or.len() > 0 {
        //         let connector = if cond_in.len() > 0 && cond_or.len() > 0 {
        //             "OR"
        //         } else {
        //             ""
        //         };
        //         format!(
        //             "{sql_prefix} AND ({} {} {}) ORDER BY TABLE_NAME",
        //             cond_in, connector, cond_or
        //         )
        //     } else {
        //         format!("{sql_prefix} ORDER BY TABLE_NAME")
        //     }
        // } else if not_in_values.len() > 0 {
        //     // let placeholders = not_in_values
        //     //     .iter()
        //     //     .enumerate()
        //     //     .map(|(i, _)| format!(":{}", i + 2))
        //     //     .collect::<Vec<_>>();

        //     // format!(
        //     //     "{sql_prefix} AND TABLE_NAME NOT IN ({}) ORDER BY TABLE_NAME",
        //     //     placeholders.join(",")
        //     // )
        //     let mut placeholders = Vec::new();
        //     let mut index = 2;
        //     for value in not_in_values {
        //         if is_regex!(value) {
        //             continue;
        //         }
        //         placeholders.push(format!(":{}", index));
        //         index += 1;
        //     }
        //     let cond_in = if placeholders.len() > 0 {
        //         format!("TABLE_NAME NOT IN ({})", placeholders.join(","))
        //     } else {
        //         String::new()
        //     };

        //     placeholders.clear();
        //     for value in not_in_values {
        //         if !is_regex!(value) {
        //             continue;
        //         }
        //         placeholders.push(format!("NOT REGEXP_LIKE (TABLE_NAME, :{})", index));
        //         index += 1;
        //     }

        //     let cond_or = if placeholders.len() > 0 {
        //         format!("{}", placeholders.join(" AND "))
        //     } else {
        //         String::new()
        //     };

        //     if cond_in.len() > 0 || cond_or.len() > 0 {
        //         let connector = if cond_in.len() > 0 && cond_or.len() > 0 {
        //             "AND"
        //         } else {
        //             ""
        //         };
        //         format!(
        //             "{sql_prefix} AND ({} {} {}) ORDER BY TABLE_NAME",
        //             cond_in, connector, cond_or
        //         )
        //     } else {
        //         format!("{sql_prefix} ORDER BY TABLE_NAME")
        //     }
        // } else {
        //     format!("{sql_prefix} ORDER BY TABLE_NAME")
        // }
        let mut sql = String::from(ALL_TABLES);
        // sql.push_str(" where OWNER = :1 ");
        // for (i, value) in values.iter().enumerate() {
        //     if value.re {
        //         sql.push_str(&format!(
        //             " and {} REGEXP_LIKE (TABLE_NAME, :{})",
        //             if value.not { "not" } else { "" },
        //             i + 2
        //         ));
        //     } else {
        //         sql.push_str(&format!(
        //             " and TABLE_NAME {} in (:{})",
        //             if value.not { "not" } else { "" },
        //             i + 2
        //         ));
        //     }
        // }

        let key = "TABLE_NAME";
        // let s = "LBK";
        let mut in_values = vec![];
        let mut not_in_values = vec![];
        let mut params = Vec::new();
        params.push(s.to_owned());

        for v in values {
            if v.re {
                continue;
            }
            params.push(v.value.clone());
            if !v.not {
                in_values.push(v.value.clone());
            } else {
                not_in_values.push(v.value.clone());
            }
        }

        let cond_in_values = if in_values.len() > 0 {
            format!(
                " and {key} in ({})",
                in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + 2))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        let cond_not_in_values = if not_in_values.len() > 0 {
            format!(
                " and {key} not in ({})",
                not_in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + in_values.len() + 2))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        // 模糊匹配
        let mut regex_likes = String::new();
        let mut not_regex_likes = String::new();
        let mut start_index = in_values.len() + not_in_values.len() + 2;
        for v in values {
            if !v.re {
                continue;
            }
            if !v.not {
                params.push(v.value.clone());
                regex_likes.push_str(&format!(" and REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }

        for v in values {
            if !v.re {
                continue;
            }
            if v.not {
                params.push(v.value.clone());
                not_regex_likes
                    .push_str(&format!(" and NOT REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }
        let sql_condition = format!(
            " where owner = :1 {cond_in_values} {cond_not_in_values} {regex_likes} {not_regex_likes} order by {key}"
        );
        sql.push_str(&sql_condition);

        (sql, params)
    }

    // 查询指定表，不分页查询
    // -- 优先级: 包含全匹配 > 包含模糊匹配 > 排除全匹配 > 排除模糊匹配
    pub fn columns(&self, s: &str, t: &str, values: &Vec<RuleValue>) -> (String, Vec<String>) {
        let mut sql = String::from(ALL_COLUMNS);
        let key = "COLUMN_NAME";
        let mut in_values = vec![];
        let mut not_in_values = vec![];
        let mut params = Vec::new();
        params.push(s.to_owned());
        params.push(t.to_owned());

        for v in values {
            if v.re {
                continue;
            }
            params.push(v.value.clone());
            if !v.not {
                in_values.push(v.value.clone());
            } else {
                not_in_values.push(v.value.clone());
            }
        }

        let cond_in_values = if in_values.len() > 0 {
            format!(
                " and {key} in ({})",
                in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + 3))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        let cond_not_in_values = if not_in_values.len() > 0 {
            format!(
                " and {key} not in ({})",
                not_in_values
                    .iter()
                    .enumerate()
                    .map(|(i, _)| format!(":{}", i + in_values.len() + 3))
                    .collect::<Vec<_>>()
                    .join(",")
            )
        } else {
            String::new()
        };

        // 模糊匹配
        let mut regex_likes = String::new();
        let mut not_regex_likes = String::new();
        let mut start_index = in_values.len() + not_in_values.len() + 3;
        for v in values {
            if !v.re {
                continue;
            }
            if !v.not {
                params.push(v.value.clone());
                regex_likes.push_str(&format!(" and REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }

        for v in values {
            if !v.re {
                continue;
            }
            if v.not {
                params.push(v.value.clone());
                not_regex_likes
                    .push_str(&format!(" and NOT REGEXP_LIKE ({key}, :{})", start_index));
                start_index += 1;
            }
        }
        let sql_condition = format!(
            " where owner = :1 and table_name = :2 {cond_in_values} {cond_not_in_values} {regex_likes} {not_regex_likes} order by COLUMN_ID"
        );
        sql.push_str(&sql_condition);

        (sql, params)
    }

    // 查询表，分页查询
    // pub fn tables_with_paging(&self) -> String {
    //     get_paging_sql(ALL_TABLES)
    // }

    pub fn count(&self, s: &str, t: &str) -> String {
        format!("select count(*) from \"{}\".\"{}\"", s, t)
    }
}

// impl DatabaseDialect for OracleDialect {
//     // 分页查询
//     fn paging<T, C>(
//         tab_name: T,
//         columns: Vec<C>,
//         order_by_keys: T,
//         offset: usize,
//         limit: usize,
//     ) -> T
//     where
//         T: Into<String>,
//         C: TableColumn,
//     {

//     }
// }

// const PAGING_TABLE_COLUMN: &str =
//     "select a.OWNER, a.TABLE_NAME, a.COLUMN_NAME, a.DATA_TYPE, a.NULLABLE from ALL_TAB_COLUMNS a";

impl OracleManager {
    pub fn new<U, P, C>(username: U, password: P, connect_string: C) -> Self
    where
        U: Into<String> + Display + Debug + Clone,
        P: Into<String> + Display + Debug + Clone,
        C: Into<String> + Display + Debug + Clone,
    {
        // 如果环境变量有oracle_home
        // std::env::set_var("ORACLE_HOME", "/Users/lbk/Downloads/instantclient_19_16");
        // std::env::set_var("DYLD_LIBRARY_PATH", "/Users/lbk/Downloads/instantclient_19_16");
        // std::env::set_var("PATH", "C:\\Users\\BK-liao\\Downloads\\instantclient_19_25");
        // std::env::set_var("NLS_LANG", "french_france.AL32UTF8");
        // std::env::set_var("NLS_TIMESTAMP_FORMAT", "YYYY-MM-DD HH24:MI:SS");
        // std::env::set_var("NLS_TIMESTAMP_TZ_FORMAT", "YYYY-MM-DD HH24:MI:SS");

        log::info!("Connecting to [{username}] => {connect_string}");
        let pool = PoolBuilder::new(username.clone(), password, connect_string.clone())
            .max_connections(16)
            .build();
        let mut pool = match pool {
            Ok(pool) => pool,
            Err(e) => {
                log::error!("{e}");
                exit(-1);
            }
        };
        // pool.set_timeout(Duration::from_secs(30)).unwrap();
        pool.set_get_mode(&pool::GetMode::Wait).unwrap();

        let dialect = OracleDialect::new();
        let om = OracleManager { pool, dialect };
        om.init();
        om
    }

    #[inline]
    fn init(&self) {
        let client_ver = Version::client().unwrap();
        let conn = self.get_conn();
        let (server_ver, _) = conn.server_version().unwrap();
        log::info!(
            "Oracle Client Version: {}, Oracle Server version: {}",
            client_ver,
            server_ver
        );
        // log::info!("Oracle Connection {:?}", conn.info().unwrap());
        match conn.query_as::<(String, String)>(NLS_SESSION_PARAMETERS, &[]) {
            Ok(rows) => {
                for row in rows {
                    let (param, value) = row.unwrap();
                    log::info!("{param}={:?}", value);
                }
            }
            Err(e) => log::error!("{e}"),
        }
    }

    // 从连接池中获取连接
    pub fn get_conn(&self) -> oracle::Connection {
        self.pool.get().expect("No idle connections")
    }

    // 全部schema
    #[allow(unused)]
    pub fn schemas(&self) -> Vec<String> {
        let rows = self
            .get_conn()
            .query_as::<String>(&self.dialect.all_schemas(), &[])
            .expect("Fetch schema failed");
        let mut schemas = Vec::new();
        for row_result in rows {
            let row_result = row_result.expect("Fetch row failed");
            schemas.push(row_result);
        }
        schemas
    }

    pub fn checked_schemas(&self, rule: &Vec<RuleValue>) -> Vec<String> {
        let now = Local::now();
        let (sql, params) = self.dialect.schemas(rule);
        log::debug!("SQL> {sql} << ~~ {:?} ", params);
        let sql_params: Vec<&dyn oracle::sql_type::ToSql> = params
            .iter()
            .map(|k| k as &dyn oracle::sql_type::ToSql)
            .collect();

        let rows = self
            .get_conn()
            .query_as::<String>(&sql, &sql_params)
            .expect("Fetch schema failed");
        let mut schemas = Vec::new();
        for row_result in rows {
            let row_result = row_result.expect("Fetch row failed");
            schemas.push(row_result);
        }

        let now = Local::now() - now;
        log::info!(
            "Schema fetched: {} ({:.3} sec)",
            schemas.len(),
            (now.num_milliseconds() as f64 / 1000.0) as f64
        );
        schemas
    }

    // #[allow(unused)]
    // pub fn get_db_tables(&self, s: &str, tabs: Vec<String>) -> Vec<String> {
    //     if tabs.len() > 0 {
    //         return self.checked_tables(s, tabs);
    //     }
    //     self.schema_tables(s)
    // }

    //
    pub fn schema_tables(&self, s: &str) -> Vec<String> {
        let now = Local::now();
        log::debug!("SQL> {ALL_TABLES_WITH_OWNER} << ~~ [{:?}]", &s,);

        let rows = self
            .get_conn()
            .query_as::<String>(&ALL_TABLES_WITH_OWNER, &[&s])
            .unwrap();

        let mut tables: Vec<String> = Vec::new();
        for row_result in rows {
            let table_name = row_result.unwrap();
            tables.push(table_name);
        }

        let now = Local::now() - now;
        log::info!(
            "Schema [{}] tables fetched: {} ({:.3} sec)",
            s,
            tables.len(),
            (now.num_milliseconds() as f64 / 1000.0) as f64
        );

        tables
    }

    // 检查表名
    pub fn checked_tables(&self, s: &str, values: &Vec<RuleValue>) -> Vec<String> {
        let now = Local::now();
        let (sql, params) = self.dialect.tables(s, values);
        // let params: Vec<&dyn oracle::sql_type::ToSql> = if in_tabs.len() > 0 {
        //     let mut bind_values = vec![s.to_owned()];
        //     for value in in_tabs.clone() {
        //         let value = value.trim();
        //         if is_regex!(value) {
        //             continue;
        //         }
        //         bind_values.push(value.to_owned());
        //     }
        //     for value in in_tabs.clone() {
        //         let value = value.trim();
        //         if !is_regex!(value) {
        //             continue;
        //         }
        //         bind_values.push(
        //             value
        //                 .trim_start_matches("/")
        //                 .trim_end_matches("/")
        //                 .trim()
        //                 .to_owned(),
        //         );
        //     }

        //     in_tabs.clear();
        //     in_tabs = bind_values;

        //     // in_tabs.insert(0, s.to_owned());
        //     log::debug!("SQL> {sql} << ~~ {:?} ", &in_tabs);
        //     in_tabs
        //         .iter()
        //         .map(|k| k as &dyn oracle::sql_type::ToSql)
        //         .collect()
        // } else if not_in_tabs.len() > 0 {
        //     let mut bind_values = vec![s.to_owned()];
        //     for value in not_in_tabs.clone() {
        //         let value = value.trim();
        //         if is_regex!(value) {
        //             continue;
        //         }
        //         bind_values.push(value.trim().to_owned());
        //     }
        //     for value in not_in_tabs.clone() {
        //         let value = value.trim();
        //         if !is_regex!(value) {
        //             continue;
        //         }
        //         bind_values.push(
        //             value
        //                 .trim_start_matches("/")
        //                 .trim_end_matches("/")
        //                 .trim()
        //                 .to_owned(),
        //         );
        //     }

        //     not_in_tabs.clear();
        //     not_in_tabs = bind_values;

        //     log::debug!("SQL> {sql} << ~~ {:?} ", &not_in_tabs);
        //     not_in_tabs
        //         .iter()
        //         .map(|k| k as &dyn oracle::sql_type::ToSql)
        //         .collect()
        // } else {
        //     log::debug!("SQL> {sql} <<");
        //     vec![]
        // };
        let sql_params: Vec<&dyn oracle::sql_type::ToSql> = params
            .iter()
            .map(|k| k as &dyn oracle::sql_type::ToSql)
            .collect();
        log::debug!("SQL> {sql} << ~~ {:?} ", &params);

        let rows = self
            .get_conn()
            .query_as::<String>(&sql, &sql_params)
            .unwrap();
        let mut tables: Vec<String> = Vec::new();
        for row_result in rows {
            let table_name = row_result.unwrap();
            tables.push(table_name);
        }

        let now = Local::now() - now;
        log::info!(
            "Schema [{}] tables fetched: {} ({:.3} sec)",
            s,
            tables.len(),
            (now.num_milliseconds() as f64 / 1000.0) as f64
        );
        tables
    }

    // 查询表数据并计算crc码
    // pub fn query_table(&self, s: &str, t: &str, page: usize, page_size: usize) -> Vec<Checksum> {
    //     // 查询表的主键、唯一索引、列
    //     log::debug!("s: {s}, t: {t}, page: {page}, page_size: {page_size}");
    //     Vec::new()
    // }

    // 统计行数
    pub fn count(&self, table: QueryTable) -> usize {
        let now = Local::now();
        let sql = self.dialect.count(&table.owner, &table.table_name);
        log::debug!("SQL> {sql} <<");
        let n = match self.get_conn().query_row_as::<usize>(&sql, &[]) {
            Ok(n) => n,
            Err(e) => {
                log::error!(
                    "Table [{}.{}] rows count failed, {e}",
                    &table.owner,
                    &table.table_name
                );
                0
            }
        };
        let now = Local::now() - now;
        log::info!(
            "Table [{}.{}] rows counted: {n} ({:.3} sec)",
            &table.owner,
            &table.table_name,
            (now.num_milliseconds() as f64 / 1000.0) as f64
        );

        n
    }

    // 查询表的所有列
    // pub fn table_cols(&self, s: &str, t: &str) -> Vec<ColumnInfo> {
    //     Vec::new()
    // }

    // pub fn tables(
    //     &self,
    //     s: &str,
    //     mut tabs: Vec<String>,
    //     page: usize,
    //     page_size: usize,
    // ) -> Vec<String> {
    //     let rows = if tabs.len() > 0 {
    //         let sql = self.dialect.tables(&tabs);
    //         tabs.insert(0, s.to_owned());
    //         log::debug!("SQL> {sql} << ~~ {:?} ", &tabs);
    //         let params: Vec<&dyn oracle::sql_type::ToSql> = tabs
    //             .iter()
    //             .map(|k| k as &dyn oracle::sql_type::ToSql)
    //             .collect();

    //         self.get_conn().query_as::<String>(&sql, &params)
    //     } else {
    //         // :1
    //         let end = page * page_size;
    //         // :2
    //         let start = (page - 1) * page_size + 1;
    //         // let schema = s.schema();
    //         let sql = self.dialect.tables_with_paging();
    //         log::debug!(">> {sql} << ~~ [{:?}, {:?}, {:?}]", &s, &end, &start);

    //         self.get_conn()
    //             .query_as::<String>(&sql, &[&end, &start, &s])
    //     }
    //     .unwrap();

    //     let mut tables: Vec<String> = Vec::new();
    //     for row_result in rows {
    //         let table_name = row_result.unwrap();
    //         tables.push(table_name);
    //     }
    //     tables
    // }

    pub fn get_tab_cols(&self, table_info: &mut TableInfo, values: &Vec<RuleValue>) {
        // let mut table_cols = Vec::new();
        let (sql, params) =
            self.dialect
                .columns(&table_info.schema_name, &table_info.table_name, values);
        let sql_params: Vec<&dyn oracle::sql_type::ToSql> = params
            .iter()
            .map(|k| k as &dyn oracle::sql_type::ToSql)
            .collect();
        log::debug!("SQL> {sql} << ~~ {:?} ", &params);

        match self
            .get_conn()
            .query_as::<QueryTableColumn>(&sql, &sql_params)
        {
            Ok(rows) => {
                for row in rows {
                    match row {
                        Ok(qtc) => {
                            let c = qtc.column_name.clone();
                            let mut ci = ColumnInfo::default();
                            ci.column_name = qtc.column_name.clone();
                            ci.data_type = qtc.data_type.clone();
                            ci.nullable = qtc.nullable;

                            ci.long = match qtc.data_type.as_str() {
                                "LONG" | "CLOB" | "NCLOB" | "XMLTYPE" | "LONG RAW" | "BLOB"
                                | "BFILE" => true,
                                _ => false,
                            };

                            if qtc.nullable {
                                table_info.nullable_cols.push(qtc.column_name);
                            }
                            ci.query_column_name = if qtc.data_type.starts_with("TIMESTAMP") {
                                format!("\"{}\"+0 as \"{}\"", c, c)
                            } else {
                                format!("\"{}\"", c)
                            };

                            // col_types.push(qtc.data_type);
                            table_info.cols.push(ci);
                        }
                        Err(e) => {
                            log::error!(
                                "Table [{},{}] column convert to QueryTableColumn failed, {}",
                                &table_info.schema_name,
                                &table_info.table_name,
                                e
                            );
                        }
                    }
                }
            }
            Err(e) => {
                log::error!(
                    "Table [{},{}] column fetch failed, {}",
                    &table_info.schema_name,
                    &table_info.table_name,
                    e
                );
            }
        }
    }

    fn query_table_pk(&self, table_info: &mut TableInfo) {
        // let mut primary_keys = Vec::new();
        log::debug!(
            "SQL> {TAB_CONSTRAINT_PK} << ~~ [{:?}, {:?}] ",
            &table_info.schema_name,
            &table_info.table_name,
        );

        match self.get_conn().query_as::<String>(
            TAB_CONSTRAINT_PK,
            &[&table_info.schema_name, &table_info.table_name],
        ) {
            Ok(rows) => {
                for row in rows {
                    match row {
                        Ok(column) => {
                            table_info.keys.push(column);
                        }
                        Err(e) => {
                            log::error!(
                                "Table [{},{}] column convert to String failed, {}",
                                &table_info.schema_name,
                                &table_info.table_name,
                                e
                            );
                        }
                    }
                }
            }
            Err(e) => {
                log::error!(
                    "Table [{},{}] PrimaryKeys fetch failed, {}",
                    &table_info.schema_name,
                    &table_info.table_name,
                    e
                );
            }
        }
        // table_info.keys = primary_keys;
    }

    fn query_table_ui(&self, table_info: &mut TableInfo) {
        log::debug!(
            "SQL> {TAB_CONSTRAINT_UI} << ~~ [{:?}, {:?}] ",
            &table_info.schema_name,
            &table_info.table_name,
        );
        match self.get_conn().query_as::<(String, String)>(
            TAB_CONSTRAINT_UI,
            &[&table_info.schema_name, &table_info.table_name],
        ) {
            Ok(rows) => {
                for row in rows {
                    match row {
                        Ok((_index_name, cols)) => {
                            // 判断是否存在可空列
                            let mut nullable = false;
                            for col in cols.split(",") {
                                if table_info.nullable_cols.contains(&col.to_owned()) {
                                    nullable = true;
                                    break;
                                }
                            }

                            if !nullable {
                                table_info.keys = cols.split(",").map(|c| c.to_owned()).collect();
                                break;
                            }
                        }
                        Err(e) => {
                            log::error!(
                                "Table [{},{}] column convert to (String, String) failed, {}",
                                &table_info.schema_name,
                                &table_info.table_name,
                                e
                            );
                        }
                    }
                }
            }
            Err(e) => {
                log::error!(
                    "Table [{},{}] UniqueKeys fetch failed, {}",
                    &table_info.schema_name,
                    &table_info.table_name,
                    e
                );
            }
        }
    }

    // 查询表的元数据
    // :: 表结构，主键、非空唯一索引
    pub fn query_table_metadata(
        &self,
        table: &QueryTable,
        rule: &Vec<RuleValue>,
        action: &Action,
    ) -> TableInfo {
        let mut table_info = TableInfo::new(table.owner.clone(), table.table_name.clone());

        match action {
            Action::Columns => {
                self.get_tab_cols(&mut table_info, rule);
            }
            Action::Primarykey => {
                self.query_table_pk(&mut table_info);
            }
            Action::NonNullableUniqueIndex => {
                self.get_tab_cols(&mut table_info, rule);
                self.query_table_ui(&mut table_info);
            }

            Action::CrcAll | Action::CrcRow | Action::Keys => {
                self.get_tab_cols(&mut table_info, rule);
                self.query_table_pk(&mut table_info);
                if table_info.keys.len() == 0 {
                    self.query_table_ui(&mut table_info);
                }
            }

            _ => {}
        }
        let col_types = table_info
            .cols
            .iter()
            .map(|c| c.data_type.to_owned())
            .collect::<Vec<String>>();
        log::debug!("TYP> {:?}", col_types);

        table_info
    }
}
